package com.openess.bigsearch.engine.BTree;

import java.io.File;
import java.io.IOException;
import java.io.RandomAccessFile;
import java.nio.ByteBuffer;
import java.nio.channels.FileChannel;

import com.openess.bigsearch.engine.BTree.util.ByteArrays;

/**
 * 此B树只是传统B树，并没有和索引进行结合，如何和索引结合还需要考虑几点问题，这些问题将在B+树版本中得以实现：1.
 * 此处的B树的节点存储的关键字没有考虑是什么数据
 * ，对于索引来说，节点内装载的应该是String类型的分词，以及分词对应的docID链表所在磁盘的偏移位置，而不是将所有数据放入节点内
 * 。2.节点合并策略没有实现
 * 。3.线程安全问题没有考虑。4.多线程建索引和多线程检索没有考虑。5.关键字和指针的压缩没有考虑。6.当将节点从磁盘加载到内存后
 * ，如果使用数组来存储这些key ，那么在插入的时候
 * ，从平均来看，每次插入都会移动一半的数据，耗时过长，因此可以考虑每个节点内部，再使用树形结构在存储节点中的key，这样可以缩短插入关键字所花费的时间
 * ，如果使用数组时间复杂度
 * ：O(K/2)，使用树：O(logK)。7.可以用使用bulk-loading，针对空树先排序后在最右节点连续插入，可节省大量插入时间。
 * 
 * @author TuTu
 * 
 */
public class BTree_new {
	private RandomAccessFile rf;
	private FileChannel cf;
	private File file;
	/**
	 * 树中节点的棵树
	 */
	private int nodeCount = 1;
	/**
	 * 整个索引文件所占的字节数
	 */
	private long fileSize;
	/**
	 * 树的根节点，始终加载到内存中
	 */
	private BTree_node rootNode;
	/**
	 * 存放头信息的字节数组
	 */
	private byte[] head;
	/**
	 * 根节点ID
	 */
	private int rootID;
	/**
	 * 关键字大小的比较器
	 */
	private BKComparable compare;

	/**
	 * 头信息占多少字节
	 */
	public static int HEADSIZE = 16;
	/**
	 * 一个节点占多少字节，包括：1.头4个字节存储的是实际key个数。2.第一个单独的NodeID
	 * 
	 */
	public static final int NODESIZE = (BTree_node.MAXKEY)
			* BTree_node.PAIRESIZE + BTree_node.HEADSIZE;

	/**
	 * 构造方法，一上来先把根节点加载到内存中，保证根节点始终在内存中，如果磁盘索引是新建的，那么先创建一个根节点
	 * 
	 * @throws IOException
	 */
	public BTree_new() throws IOException {
		file = new File("E://index");
		rf = new RandomAccessFile(file, "rw");
		cf = rf.getChannel();
		if (cf.size() == 0) {
			writeHead();
			createRootNode();
		} else {
			init();
		}
	}

	/**
	 * 树的初始化，读取磁盘，将变量赋值
	 */
	private void init() {
		head = new byte[HEADSIZE];
		fileSize = ByteArrays.getLong(head, 0);
		nodeCount = ByteArrays.getInt(head, 8);
		rootID = ByteArrays.getInt(head, 12);
		rootNode = getNode(rootID);
		compare = new defaultBKCompare();
	}

	/**
	 * 在文件内写入头数据信息
	 */
	private void writeHead() {
		head = new byte[HEADSIZE];
		ByteArrays.putLong(fileSize, head, 0);
		ByteArrays.putInt(nodeCount, head, 8);
		ByteArrays.putInt(rootID, head, 12);
	}

	/**
	 * 创建根节点
	 */
	private void createRootNode() {
		this.rootNode = new BTree_node(nodeCount++);
		this.rootID = rootNode.nodeID;
		rootNode.leaf = true;
		rootNode.write();
	}

	/**
	 * 插入字符类型的关键字
	 * 
	 * @param key
	 */
	public void insert(String str) {
		// 不能使用默认的比较器
		compare = new StringBKCompare();
		byte[] strKey = createStrKey(str);
		insertFromRoot(strKey);
	}

	/**
	 * 创建一个专门String类型的关键字，前两个字节存储了字符个数，用于以后读取
	 * 
	 * @param str
	 * @return
	 */
	private byte[] createStrKey(String str) {
		// 如果字符串的长度超过了10，那么对其进行截断，保证key值不超出界限
		if (str.length() > BTree_node.KEYSIZE) {
			str = str.substring(0, 10);
		}
		int size = str.length() * 2 + BTree_node.STRSIZE;
		byte[] dest = new byte[size];
		ByteArrays.putShort((short) str.length(), dest, 0);
		ByteArrays.putChars(str, dest, 2);
		return dest;
	}

	/**
	 * 关键字从根节点进行插入。
	 * 关键字插入过程理解：一开始只有一个空的根节点，并且为叶子节点。因为插入操作只会发生在叶子节点上，因此直到根节点被插满，这时候要增加新的节点层
	 * ，第一个非页节点诞生
	 * ，也就是新的根节点。在一颗有N个节点的多层树中，进行插入，同样遵循：只能在叶节点上进行插入。每当叶结点满时，并不是叶节点升为内节点
	 * ，而是叶结点的其中一个值升入内节点
	 * ，并把1个叶子节点拆成2个叶子节点。因此实际上从一开始是叶子的节点，即使插入了N多的key，其最终还是叶结点，永远无法变成内节点
	 * ，叶结点的增加只会发生在分裂时
	 * ，分裂出一个兄弟节点，其也是叶节点。因为所有的插入操作都只发生在叶节点上，因此内结点的key个数增加，就完全依靠其孩子节点分裂一次
	 * ，内节点就增加一个溢出key。
	 * 
	 * 根据这个原理，如何设置节点是否为叶节点，有三个地方：1.在创建新树的根节点的时候leaf=true。2.根节点分裂的时候将新的根leaf=
	 * false。3.普通节点分裂的时候，node.leaf = brotherNode.leaf。
	 * 
	 * @param key
	 */
	public void insertFromRoot(byte[] key) {
		if (rootNode.isFull()) {
			splitRoot();
		}
		insert(key, rootNode);
	}

	/**
	 * 从指定的子树节点开始进行插入操作，这个过程不能单独运行，因为没有进行根节点的分裂操作，运行前提：根节点已进行过分裂。此处没有使用尾递归，
	 * 而是使用循环，可以提高效率。
	 * 
	 * @param key
	 * @param node
	 * 
	 */
	private void insert(byte[] key, BTree_node node) {
		while (true) {
			int result = node.search(key);
			// 如果节点内已有相同的key，那么不执行插入操作
			if (result >= 0) {
				return;
			} else {
				result = -result - 1;
				// 插入操作只发生在叶子节点上
				if (node.isLeaf()) {
					node.insertKeyPaire(key, -1, result);
					return;
				} else {
					int childID = node.getChildID(result - 1);
					BTree_node childNode = getNode(childID);
					// 如果节点中key满，进行分裂
					if (childNode.isFull()) {
						byte[] key1 = splitNode(node, childNode, result);
						// 和分裂后插入的节点的关键字进行比较，决定最终应该进入哪个孩子节点
						if (compare.compare(key1, key, 0, BTree_node.KEYSIZE) > 0) {
							result++;
						}
						// 写回磁盘
						node.write();
						node = childNode;
					}
				}
			}
		}
	}

	/**
	 * 如果根节点满了，那么需要将根节点进行分裂，并重新设置根节点
	 */
	private void splitRoot() {
		// 新的根节点
		BTree_node node = new BTree_node(nodeCount++);
		// 原先的根节点变成左孩子
		int leftChildID = rootNode.nodeID;
		// 拆分根节点
		byte[] key = rootNode.splitNode();
		// 往新的根节点内放入数据
		ByteArrays.put(key, 0, node.data, BTree_node.HEADSIZE
				+ BTree_node.IDSIZE, BTree_node.PAIRESIZE);
		ByteArrays.putInt(leftChildID, node.data, BTree_node.HEADSIZE);
		// 在内存中替换掉旧的根节点
		setRoot(node);
	}

	/**
	 * 将节点分裂，并完成将溢出的关键字插入父节点的过程
	 * 
	 * @param parent
	 *            被分裂节点的父节点
	 * @param node
	 *            被分裂的节点
	 * @param insertIndex
	 *            在父节点的什么位置将溢出值插入
	 * @return 插入到父节点中的关键字
	 */
	private byte[] splitNode(BTree_node parent, BTree_node node, int insertIndex) {
		byte[] paire = node.splitNode();
		parent.insertKeyPaire(paire, insertIndex);
		return paire;
	}

	/**
	 * 将节点设置为树的根节点
	 * 
	 * @param node
	 *            要被设成根节点
	 */
	private void setRoot(BTree_node node) {
		this.rootID = node.nodeID;
		this.rootNode = node;
	}

	/**
	 * 搜索字符类关键字
	 * 
	 * @param str
	 * @return
	 */
	public String searchKey(String str) {
		// 不能使用默认比较器
		compare = new StringBKCompare();
		byte[] strKey = createStrKey(str);
		byte[] result = searchKeyTree(strKey);
		return readStrKey(result);
	}

	/**
	 * 
	 * @param strKey
	 * @return
	 */
	private String readStrKey(byte[] strKey) {
		int length = ByteArrays.getShort(strKey, 0);
		return ByteArrays.getString(strKey, BTree_node.STRSIZE, length);
	}

	/**
	 * 从整颗树中检索关键字
	 * 
	 * @param key
	 * @return
	 */
	public byte[] searchKeyTree(byte[] key) {
		return searchKey(key, rootNode);
	}

	/**
	 * 从指定的子树中检索关键字
	 * 
	 * @param key
	 *            待搜索的关键字
	 * @param node
	 *            搜索子树的根节点
	 * @return
	 */
	private byte[] searchKey(byte[] key, BTree_node node) {
		if (node == null || key == null) {
			return null;
		}
		int result;
		// 此方法还可以用尾递归来实现，但用循环的效率更高
		while (true) {
			result = node.search(key);
			// 不管本节点是否为叶子，先行判断是否找到了关键字
			if (result < 0) {
				// 因为在节点中childID的数量比key多一个，且两者都是从零开始计数，因此当childIndex =
				// 5时，对应的keyIndex=4
				int childIndex = -result - 1;
				// 如果在本节点内没有找到关键字，并且本节点没有孩子了，那么可以说明搜索结束，树中没有指定的关键字
				if (node.isLeaf()) {
					return null;
				} else {
					int nodeID = node.getChildID(childIndex);
					node = getNode(nodeID);
				}
			} else {
				return node.getKey(result);
			}
		}
	}

	/**
	 * 从磁盘中拿到节点
	 * 
	 * @param nodeID
	 * @return
	 */
	private BTree_node getNode(int nodeID) {
		BTree_node node = new BTree_node(nodeID);
		int nodeOffset = nodeOffset(nodeID);
		ByteBuffer buf = ByteBuffer.wrap(node.data);
		try {
			cf.read(buf, nodeOffset);
		} catch (IOException e) {
			e.printStackTrace();
		}
		return node;
	}

	/**
	 * 在整个磁盘中寻找指定节点的偏移量
	 * 
	 * @param nodeID
	 *            节点的ID
	 * @return 偏移量
	 */
	private int nodeOffset(int nodeID) {
		return HEADSIZE + nodeID * NODESIZE;
	}

	/**
	 * 关闭B树的所有连接
	 * 
	 * @throws IOException
	 */
	public void close() throws IOException {
		rf.close();
		cf.close();
	}

	private class BTree_node implements Comparable<BTree_node> {
		/**
		 * 一个节点内，key的总个数上限，此值对所有节点都是固定的大小
		 */
		public static final int MAXKEY = 10;
		/**
		 * 一个关键字所占字节大小
		 */
		public static final int KEYSIZE = 22;
		/**
		 * 如果关键字是String类型的，那么关键字的前2个字节用于存储字符的个数，用short类型
		 */
		public static final int STRSIZE = 2;
		/**
		 * 一个存储在节点内的一个ID所占的字节数
		 */
		public static final int IDSIZE = 4;
		/**
		 * 节点内除了存储key,还需要存储childNodeID指针，因此组成(key,childNodeID)，一个k/v所占字节数
		 */
		public static final int PAIRESIZE = KEYSIZE + IDSIZE;
		/**
		 * 头信息所占的字节，头信息只存储关键字个数
		 */
		public static final int HEADSIZE = 4;
		/**
		 * 节点的ID号
		 */
		private int nodeID;
		/**
		 * 当前节点内的关键字个数
		 */
		private int keyCount = 1;

		/**
		 * 存放节点内所有数据的数组，最后将数组写入磁盘
		 */
		private byte[] data;
		/**
		 * 判断是否为叶子节点
		 */
		private boolean leaf = false;

		/**
		 * 构造方法，新建一个节点
		 * 
		 * @param nodeID
		 */
		public BTree_node(int nodeID) {
			this.nodeID = nodeID;
			this.data = new byte[NODESIZE];
		}

		/**
		 * 拿到节点中指定关键字的值
		 * 
		 * @param keyIndex
		 *            节点中第几个关键字，从零开始
		 * @return 关键字值的字节数组
		 */
		public byte[] getKey(int keyIndex) {
			int offset = keyOffset(keyIndex);
			return ByteArrays.get(data, offset, KEYSIZE);
		}

		/**
		 * 关键字在节点内的偏移量
		 * 
		 * @param keyID
		 *            节点的第几个关键字,从零开始。因为在节点中childID的数量比key多一个，且两者都是从零开始计数，
		 *            因此当childIndex =5时，对应的keyIndex=4
		 * @return
		 */
		private int keyOffset(int keyIndex) {
			return HEADSIZE + IDSIZE + keyIndex * PAIRESIZE;
		}

		/**
		 * 拿到当前节点中保存的孩子节点ID
		 * 
		 * @param childIndex
		 * @return
		 */
		public int getChildID(int childIndex) {
			int offset = childOffset(childIndex);
			int result = ByteArrays.getInt(this.data, offset);
			return result;
		}

		/**
		 * 孩子节点指针在父节点内的偏移量
		 * 
		 * @param childIndex
		 *            此孩子节点指针在父节点内是第几个，从零开始，数量比key的总数多一个
		 * @return
		 */
		private int childOffset(int childIndex) {
			return HEADSIZE + childIndex * PAIRESIZE;
		}

		/**
		 * 仅在本节点内搜索指定的key
		 * 
		 * @param key
		 *            要查找的key
		 * 
		 * @return 
		 *         返回keyIndex，如果返回值大于零，说明查找到指定的关键字。如果返回值小于零，则说明没有找到指定的key，返回值为return
		 *         = -low -1， 根据此值可以计算出应该把关键字插在哪里，只需要反向计算low = -return
		 *         -1，即可拿到low的值
		 *         。根据折半查找，如果没有找到结果，那么折半结束后，low的值即是第一个比key大的值，也就是按顺序插入
		 *         ，key应该插在的地方(keyIndex)
		 */
		public int search(byte[] key) {
			int low = 0;
			int high = keyCount;
			while (low <= high) {
				int mid = (low + high) / 2;
				int midoffset = keyOffset(mid);
				int result = compare.compare(key, data, midoffset, KEYSIZE);
				if (result == 0) {
					return mid;
				} else if (result > 0) {
					low = mid + 1;
				} else {
					high = mid - 1;
				}
			}
			return -low - 1;
		}

		/**
		 * 当节点内的关键字满了时，进行的分裂操作，只是进行了分裂，并没有将溢出key,nodeID对插入到父节点中
		 * 
		 * @return 待被插入到父节点的溢出key,nodeID对
		 */
		public byte[] splitNode() {
			BTree_node newNode = new BTree_node(nodeCount++);
			int splitOffset = keyOffset(MAXKEY / 2);
			// 将分裂后的右半部分数据放入到新节点当中
			ByteArrays.put(data, splitOffset + KEYSIZE, newNode.data, 4,
					NODESIZE - splitOffset);
			// 将当前节点的leaf属性传递给分裂的兄弟节点
			newNode.leaf = this.leaf;
			// 清除当前节点的右半部分
			ByteArrays.clear(this.data, splitOffset, NODESIZE - splitOffset);
			// 拿到中间key
			byte[] midKey = ByteArrays.get(data, splitOffset, KEYSIZE);
			// 组合成key,nodeID对
			byte[] midKeyPaire = new byte[PAIRESIZE];
			ByteArrays.put(midKey, 0, midKeyPaire, 0, KEYSIZE);
			ByteArrays.putInt(newNode.nodeID, midKeyPaire, KEYSIZE);
			// 新创建的节点和本分裂节点都写回磁盘
			newNode.write();
			this.write();
			return midKeyPaire;
		}

		/**
		 * 在节点中插入(key,nodeID)对,此方法不判断key是否达到上限，应该由使用者(BTree_new)来判断
		 * 
		 * @param key
		 * @param nodeID
		 * @param keyIndex
		 */
		public void insertKeyPaire(byte[] key, int nodeID, int keyIndex) {
			byte[] paire = createPaire(key, nodeID);
			insertKeyPaire(paire, keyIndex);
		}

		/**
		 * 在节点中插入(key,nodeID)对,此方法不判断key是否达到上限，应该由使用者(BTree_new)来判断
		 * 
		 * @param paire
		 * @param keyIndex
		 */
		public void insertKeyPaire(byte[] paire, int keyIndex) {
			int keyOffset = keyOffset(keyIndex);
			// 如果是尾插入key，那么不需要移动
			if (keyIndex <= keyCount) {
				int currentSize = keyOffset(keyCount + 1);
				ByteArrays.move(this.data, keyOffset, keyOffset + PAIRESIZE,
						currentSize - keyOffset + 1);
			}
			ByteArrays.put(paire, 0, data, keyOffset, PAIRESIZE);
			this.keyCount++;
		}

		/**
		 * 将一个key和nodeID组合成一个paire字节数组
		 * 
		 * @param key
		 * @param nodeID
		 * @return 返回paire字节数组
		 */
		private byte[] createPaire(byte[] key, int nodeID) {
			byte[] paire = new byte[PAIRESIZE];
			ByteArrays.put(key, 0, paire, 0, KEYSIZE);
			ByteArrays.putInt(nodeID, paire, KEYSIZE);
			return paire;
		}

		/**
		 * 已经使用完此节点，将其写回磁盘
		 */
		private void write() {
			int offset = nodeOffset(nodeID);
			ByteBuffer buf = ByteBuffer.wrap(data);
			try {
				cf.write(buf, offset);
			} catch (IOException e) {
				e.printStackTrace();
			}
		}

		/**
		 * 判断是否为叶子节点
		 * 
		 * @return 叶子节点为true
		 */
		private boolean isLeaf() {
			return leaf;
		}

		/**
		 * 判断节点内key的个数是否已满
		 * 
		 * @return
		 */
		private boolean isFull() {
			return keyCount == MAXKEY;
		}

		@Override
		public int compareTo(BTree_node o) {
			return this.nodeID - o.nodeID;
		}
	}
}
